package com.cyw.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.cyw.common.core.domain.entity.SysUser;
import com.cyw.common.utils.StringUtils;
import com.cyw.socket.entity.LoveRelation;
import com.cyw.socket.entity.Notify;
import com.cyw.socket.mapper.LoveMapper;
import com.cyw.socket.mapper.NotifyMapper;
import com.cyw.system.mapper.SysUserMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @Author: Chenyw
 * @Date: 2022/11/10/9:49
 */
@Component
@Slf4j
public class KafkaConsumer {
    @Autowired
    private NotifyMapper notifyMapper;

    @Autowired
    private SysUserMapper sysUserMapper;

    @Autowired
    private LoveMapper loveMapper;

    @KafkaListener(topics = { "moments_topic" })
    public void onMessageMoments(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.info("消费信息 topic: {}, partition: {}, value: {}", record.topic(), record.partition(), record.value());
    }

    @KafkaListener(topics = { "user_topic" })
    public void onMessageUser(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.info("消费信息 topic: {}, partition: {}, value: {}", record.topic(), record.partition(), record.value());

        /**
         * type :1 ,loginUserId ,userId
         * type: 1:关注
         */
        if (StringUtils.isNotEmpty(record.value().toString())){
            JSONObject jsonObject = JSON.parseObject(record.value().toString());
            String type = (String) jsonObject.get("type");
            if ("1".equals(type)){
                log.info("{}","关注的消息");
                pushNotifyTask(Long.parseLong((String) jsonObject.get("loginUserId")),Long.parseLong((String) jsonObject.get("userId")));
                saveLoveRelation(Long.parseLong((String) jsonObject.get("loginUserId")),Long.parseLong((String) jsonObject.get("userId")));
            }

            if ("2".equals(type)){
                log.info("{}","取关的消息");
                /**
                 * TODO:
                 */

            }
        }
    }

    /**
     * 给被关注人推送消息
     */
    private Boolean pushNotifyTask(Long loginUserId , Long userId){
        SysUser sysUser = sysUserMapper.selectUserById(loginUserId);
        Notify notify = new Notify();
        notify.setIsRead("0");
        notify.setNotifyContent("用户"+ sysUser.getNickName()+"关注了你");
        notify.setNotifyTime(new Date());
        notify.setUserId(userId);
        return notifyMapper.insert(notify)>0;
    }

    /**
     * 存关注的关系(谁关注谁(id))---此mysql数据库表备用
     */
    private void saveLoveRelation(Long loginUserId , Long userId){
        LoveRelation loveRelation = new LoveRelation();
        loveRelation.setCreateTime(new Date());
        loveRelation.setUserId(loginUserId);
        loveRelation.setBeLikedUserId(userId);
        loveMapper.insert(loveRelation);
    }
}











